package com.gaofans.nrpc.client.handler;

import com.gaofans.nrpc.common.context.ContextResponse;
import com.gaofans.nrpc.common.processor.EventListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.concurrent.EventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author GaoFans
 */
@ChannelHandler.Sharable
public class RpcResponseHandler extends SimpleChannelInboundHandler<ContextResponse> {

    public final ResponseEventProcessor processor;

    public RpcResponseHandler(ResponseEventProcessor processor) {
        this.processor = processor;
    }

    private final static Logger LOG = LoggerFactory.getLogger(RpcResponseHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ContextResponse response) throws Exception {
        long id = response.getSequenceId();
        EventExecutor executor = response.executor();
        executor.submit(() -> {
            EventListener<Object> listener = processor.unRegister(id);
            if (listener != null){
                //出现异常
                if (!response.isSuccess()){
                    listener.processError(new RuntimeException(response.getError()));
                    //数据流
                }else if (response.isStream() && !response.isEnd()){
                    listener.onDataChunk(response.getResult());
                    processor.register(id, listener);
                    //数据流结束
                }else if (response.isStream() && response.isEnd()){
                    listener.processComplete();
                    //单数据
                }else {
                    listener.onDataChunk(response.getResult());
                    listener.processComplete();
                }
            }else {
                LOG.warn("未找到response-{}监听器",response.getSequenceId());
            }
        });
    }
}
